package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.DimJoinFunction;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @author Felix
 * @date 2024/7/01
 * 旁路缓存关联维度的模板类
 */
public abstract class DimMapFunction<T> extends RichMapFunction<T, T> implements DimJoinFunction<T> {
    private Connection hbaseConn;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hbaseConn);
        RedisUtil.closeJedis(jedis);
    }
    @Override
    public T map(T obj) throws Exception {
        //根据流中对象获取要关联的维度主键
        String rowKey = getRowKey(obj);
        //根据维度的主键到Redis中获取对应的维度数据
        JSONObject dimJsonObj = RedisUtil.readDim(jedis, getTableName(), rowKey);
        if(dimJsonObj != null){
            //如果从Redis中找到了要关联的维度，直接将结果进行返回(缓存命中)
            System.out.println("~~~从Redis中获取到了"+getTableName()+"表的"+rowKey+"维度~~~");
        }else{
            //如果从Redis中找到了要关联的维度，发送请求到HBase中进行查询
            dimJsonObj = HBaseUtil.getRow(hbaseConn, Constant.HBASE_NAMESPACE,getTableName(),rowKey, JSONObject.class);
            if(dimJsonObj != null){
                // 将查询出来的维度放到Redis中缓存起来，方便下次查询使用
                System.out.println("~~~从HBase中获取到了"+getTableName()+"表的"+rowKey+"维度~~~");
                RedisUtil.writeDim(jedis,getTableName(),rowKey,dimJsonObj);
            }else{
                System.out.println("~~~要查询的"+getTableName()+"表的"+rowKey+"维度不存在~~~");
            }
        }
        //将维度属性补充到流中对象上
        if(dimJsonObj != null){
            addDims(obj,dimJsonObj);
        }
        return obj;
    }
}
